package com.flink.conn;

import com.alibaba.fastjson.JSONObject;
import com.flink.elasticsearch.ElasticSearchUtils;
import com.flink.entity.LogInfoDTO;
import com.flink.mapper.LogInfoMapper;
import com.flink.unit.Constant;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.ibatis.session.SqlSession;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;

/**
 * @author Administrator
 * @date 2020/7/4
 */
public class SinkLogToES extends RichSinkFunction<List<LogInfoDTO>> {
    String table;
    String topic;

    public SinkLogToES(String table, String topic) {
        this.table = table;
        this.topic = topic;
    }

    private static final Logger LOG = LoggerFactory.getLogger(SinkLogToES.class);

    @Override
    public void invoke(List<LogInfoDTO> value, Context context) throws Exception {
        for (LogInfoDTO logInfoDTO : value) {
            addES(logInfoDTO);
        }
    }

    public void addES(LogInfoDTO logInfoDTO) throws IOException {
        Constant constant = new Constant();
        RestHighLevelClient client = ElasticSearchUtils.getClient();
        //1.	构建IndexRequest对象，用来描述ES发起请求的数据。
        IndexRequest indexRequest = new IndexRequest(constant.es_index, constant.es_type);
        //2.	设置文档ID。
        indexRequest.id(logInfoDTO.getLogID());
        //3.	使用FastJSON将实体类对象转换为JSON。
        String json = JSONObject.toJSONString(logInfoDTO);
        //4.	使用IndexRequest.source方法设置文档数据，并设置请求的数据为JSON格式。
        indexRequest.source(json, XContentType.JSON);
        //5.	使用ES High level client调用index方法发起请求，将一个文档添加到索引中。
        IndexResponse index = client.index(indexRequest, RequestOptions.DEFAULT);
        System.out.println(JSONObject.toJSONString(index));
    }
}
